package com.rtf.framework.es.service;

import com.rtf.framework.common.json.JsonMapper;
import com.rtf.framework.common.util.UtilCollection;
import com.rtf.framework.common.util.UtilMap;
import com.rtf.framework.common.util.UtilString;
import com.rtf.framework.es.common.*;
import com.rtf.framework.es.config.AppEsHystrixConfig;
import com.rtf.framework.es.exception.*;
import com.google.common.collect.Lists;
import com.netflix.hystrix.HystrixCommand;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.*;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.GetIndexResponse;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.metrics.ValueCount;
import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.SortBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * ElasticSearch操作服务
 * @Author : lwq
 * @Date : 2021-06-02
 * @Modified By
 */
@Slf4j
@Service
public class AppHystrixElasticSearchServiceImpl<T extends AppEsEntity> implements AppElasticSearchService<T>{

    private RestHighLevelClient restHighLevelClient;
    @Autowired
    private AppEsHystrixConfig esHystrixConfig;
    @Autowired
    private AppElasticSearchService appElasticSearchService;

    public AppHystrixElasticSearchServiceImpl(RestHighLevelClient restHighLevelClient) {
        this.restHighLevelClient = restHighLevelClient;
    }

    /**
     * 获取hystrixSetter
     *
     * @param commandName
     * @return
     */
    private HystrixCommand.Setter getHystrixSetter(String commandName) {
        return esHystrixConfig.getESHystrixSetter(commandName);
    }

    /**
     * 创建索引，新版ES插入数据时自动创建
     *
     * @param index      索引
     * @param shardNum   主分片数量
     * @param replicaNum 副本数量
     * @return
     */
    @Override
    public boolean createIndex(String index, Map<String,?> mapping, Integer shardNum, Integer replicaNum) {
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("createIndex")){
            @Override
            protected Boolean run() throws Exception {
                return appElasticSearchService.createIndex(index, mapping, shardNum, replicaNum);
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("创建索引失败");
            }
        };
        return command.execute();
    }

    /**
     * 获取索引
     *
     * @throws IOException
     */
    @Override
    public GetIndexResponse getIndex(String index) {
        HystrixCommand<GetIndexResponse> command = new HystrixCommand<GetIndexResponse>(getHystrixSetter("getIndex")){
            @Override
            protected GetIndexResponse run() throws Exception {
                return appElasticSearchService.getIndex(index);
            }
            @Override
            protected GetIndexResponse getFallback() {
                throw new AppEsFallbackException("获取索引信息失败");
            }
        };
        return command.execute();
    }

    /**
     * 判断是否拥有某个索引
     *
     */
    @Override
    public Boolean existsIndex(String index) {
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("existsIndex")){
            @Override
            protected Boolean run() throws Exception {
                return appElasticSearchService.existsIndex(index);
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("判断索引是否存在失败");
            }
        };
        return command.execute();
    }

    /**
     * 删除索引
     */
    @Override
    public boolean deleteIndex(String index) {
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("deleteIndex")){
            @Override
            protected Boolean run() throws Exception {
                return appElasticSearchService.existsIndex(index);
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("删除索引失败");
            }
        };
        return command.execute();
    }

    /**
     * 更新单条数据
     */
    @Override
    public void updateDocument(String index, T esEntity) {
        Assert.isTrue(esEntity != null && UtilString.isNotBlank(esEntity.getId()), "id不能为null");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("updateDocument")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.updateDocument(index, esEntity);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("删除索引失败");
            }
        };
        command.execute();
    }

    /**
     * 更新单条数据
     */
    @Override
    public void updateDocument(String index, Map<String, Object> params) {
        Assert.isTrue(params != null && UtilMap.getString(params, "id") != null, "id不能为null");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("updateDocument")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.updateDocument(index, params);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("更新单条数据失败");
            }
        };
        command.execute();
    }

    /**
     * 批量更新
     *
     * @param index
     * @param list
     * @return
     * @throws IOException
     */
    @Override
    public void updateBatch(String index, List<Map<String, Object>> list) {
        Assert.isTrue(UtilCollection.isNotEmpty(list), "数据不能为空");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("updateBatch")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.updateBatch(index, list);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("批量更新失败");
            }
        };
        command.execute();
    }

    /**
     * 批量更新数据
     *
     * @param index
     * @param list
     * @return
     */
    @Override
    public void updateBatchDocument(String index, List<T> list) {
        Assert.isTrue(UtilCollection.isNotEmpty(list), "数据不能为空");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("updateBatchDocument")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.updateBatchDocument(index, list);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("批量更新失败");
            }
        };
        command.execute();
    }

    /**
     * 批量更新
     *
     * @param index
     * @param list
     * @return BulkResponse
     * @throws IOException
     */
    @Override
    public void updateBatchByScript(String index, List<AppEsScriptInfo> list) {
        Assert.isTrue(UtilCollection.isNotEmpty(list), "数据不能为空");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("updateBatchByScript")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.updateBatchByScript(index, list);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("批量更新失败");
            }
        };
        command.execute();
    }

    /**
     * 批量更新
     *
     * @param index
     * @param queryBuilder  查询条件
     * @param batchSize     批量更改大小
     */
    @Override
    public void updateByQuery(String index, QueryBuilder queryBuilder, Integer batchSize, Map<String, Object> document) {
        Assert.isTrue(UtilMap.isNotEmpty(document), "更新字段不能为空");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("updateByQuery1")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.updateByQuery(index, queryBuilder, batchSize, document);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("批量更新失败");
            }
        };
        command.execute();
    }

    /**
     * 批量更新
     *
     * @param index
     * @param queryBuilder 查询条件
     * @param batchSize    批量更新大小
     * @throws IOException
     */
    @Override
    public void updateByQuery(String index, QueryBuilder queryBuilder, Integer batchSize, Script script) {
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("updateByQuery2")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.updateByQuery(index, queryBuilder, batchSize, script);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("批量更新失败");
            }
        };
        command.execute();
    }

    /**
     * 保存信息
     *
     * @param index      索引名称
     * @param esEntities
     * @return IndexResponse
     * @throws Exception
     * @author lwq
     */
    @Override
    public void insertDocument(String index, T esEntities) {
        Assert.isTrue(esEntities != null && esEntities.getId() != null, "保存信息不能为空");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("insertDocument")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.insertDocument(index, esEntities);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("保存信息失败");
            }
        };
        command.execute();
    }


    /**
     * 批量插入数据
     */
    @Override
    public void insertBatchDocument(String index, List<T> esEntities) {
        Assert.isTrue(UtilCollection.isNotEmpty(esEntities), "数据不能为空");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("insertBatchDocument")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.insertBatchDocument(index, esEntities);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("批量插入数据失败");
            }
        };
        command.execute();
    }

    /**
     * 通过ID获取数据
     *
     * @param index         索引，类似数据库
     * @param id            数据ID
     * @param includeFields 需要显示的字段
     * @param excludeFields 不需要显示的字段
     * @return Map
     * @author lwq
     */
    @Override
    public Map<String, Object> getDataById(String index, String id, String[] includeFields, String[] excludeFields) {
        Assert.isTrue(UtilString.isNotEmpty(id), "id不能为空");
        final String[] includeFields1 = includeFields;
        final String[] excludeFields1 = excludeFields;
        HystrixCommand<Map<String, Object>> command = new HystrixCommand<Map<String, Object>>(getHystrixSetter("getDataById")){
            @Override
            protected Map<String, Object> run() throws Exception {
                return appElasticSearchService.getDataById(index, id, includeFields1, excludeFields1);
            }
            @Override
            protected Map<String, Object> getFallback() {
                throw new AppEsFallbackException("通过ID获取数据失败");
            }
        };
        return command.execute();
    }

    /**
     * 通过ID获取数据
     *
     * @param index 索引，类似数据库
     * @param id    数据ID
     * @return
     * @author lwq
     */
    @Override
    public Map<String, Object> getDataById(String index, String id) {
        return getDataById(index, id, null, null);
    }

    /**
     * 通过ID获取数据
     *
     * @param index         索引，类似数据库
     * @param id            数据ID
     * @param includeFields 需要显示的字段
     * @return
     */
    @Override
    public Map<String, Object> getDataByIdIncludeFields(String index, String id, String[] includeFields) {
        return getDataById(index, id, includeFields, null);

    }

    /**
     * 通过ID获取数据
     *
     * @param index         索引，类似数据库
     * @param id            数据ID
     * @param excludeFields 不需要显示的字段
     * @return
     */
    @Override
    public Map<String, Object> getDataByIdExcludeFields(String index, String id, String[] excludeFields) {
        return getDataById(index, id, null, excludeFields);
    }

    /**
     * 根据id获取详情
     *
     * @param index     索引名称
     * @param id        id
     * @param tClass
     * @return
     * @throws Exception
     */
    @Override
    public T getEntityById(String index, String id, Class<T> tClass) {
        Assert.isTrue(UtilString.isNotEmpty(id), "id不能为空");
        HystrixCommand<T> command = new HystrixCommand<T>(getHystrixSetter("getEntityById")){
            @Override
            protected T run() throws Exception {
                try {
                    GetRequest request = new GetRequest(index, id);
                    GetResponse response = restHighLevelClient.get(request, RequestOptions.DEFAULT);
                    if (response.isExists()) {
                        String dataStar = response.getSourceAsString();
                        return JsonMapper.parseObject(dataStar, tClass);
                    }
                    return null;
                } catch (Exception e) {
                    log.error("获取详情信息失败");
                    throw e;
                }
            }
            @Override
            protected T getFallback() {
                throw new AppEsFallbackException("通过ID获取数据失败");
            }
        };
        return command.execute();
    }

    /**
     * 根据查询方式获取某个索引下的文档个数
     *
     * @param index
     * @param queryBuilder
     * @return
     */
    @Override
    public long queryCount(String index, QueryBuilder queryBuilder) {
        HystrixCommand<Long> command = new HystrixCommand<Long>(getHystrixSetter("queryCount")){
            @Override
            protected Long run() throws Exception {
                return appElasticSearchService.queryCount(index, queryBuilder);
            }
            @Override
            protected Long getFallback() {
                throw new AppEsFallbackException("根据查询方式获取某个索引下的文档个数失败");
            }
        };
        return command.execute();
    }

    /**
     * 查看文档是否存在
     *
     * @param id
     * @return
     * @throws IOException
     */
    @Override
    public Boolean existsDocument(String index, String id) {
        Assert.isTrue(UtilString.isNotEmpty(id), "id不能为空");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("existsDocument")){
            @Override
            protected Boolean run() throws Exception {
                return appElasticSearchService.existsDocument(index, id);
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("查看文档是否存在失败");
            }
        };
        return command.execute();
    }

    /**
     * 多条件组合查询
     * @param index         索引名称
     * @param queryBuilder  查询条件
     * @param resultClass   返回结果类型
     * @param size          查询数量
     * @return
     * @throws IOException
     */
    @Override
    public List<T> searchDocument(String index, QueryBuilder queryBuilder, Class<T> resultClass, int size) {
        HystrixCommand<List<T>> command = new HystrixCommand<List<T>>(getHystrixSetter("searchDocument")){
            @Override
            protected List<T> run() throws Exception {
                return appElasticSearchService.searchDocument(index, queryBuilder, resultClass, size);
            }
            @Override
            protected List<T> getFallback() {
                throw new AppEsFallbackException("多条件组合查询失败");
            }
        };
        return command.execute();
    }

    /**
     * scroll 查询
     * @param index             索引名
     * @param scroll            滚动查询有效时间设置
     * @param queryBuilder      查询信息
     * @param size              查询条数
     * @param sortBuilders      排序条件
     * @param resultClass       返回类型
     * @return
     */
    @Override
    public AppEsScrollPage<T> queryScrollPage(String index, Scroll scroll, QueryBuilder queryBuilder, int size, List<SortBuilder> sortBuilders, Class<T> resultClass) {
        HystrixCommand<AppEsScrollPage<T>> command = new HystrixCommand<AppEsScrollPage<T>>(getHystrixSetter("queryScrollPage")){
            @Override
            protected AppEsScrollPage<T> run() throws Exception {
                return appElasticSearchService.queryScrollPage(index, scroll, queryBuilder, size, sortBuilders, resultClass);
            }
            @Override
            protected AppEsScrollPage<T> getFallback() {
                throw new AppEsFallbackException("scroll 查询失败");
            }
        };
        return command.execute();
    }


    /**
     *
     * @param scroll
     * @param scrollId
     * @param size
     * @param resultClass
     * @return
     */
    @Override
    public AppEsScrollPage<T> queryScrollPage(Scroll scroll, String scrollId, int size, Class<T> resultClass) {
        String scrollId1 = scrollId;
        HystrixCommand<AppEsScrollPage<T>> command = new HystrixCommand<AppEsScrollPage<T>>(getHystrixSetter("queryScrollPage1")){
            @Override
            protected AppEsScrollPage<T> run() throws Exception {
                return appElasticSearchService.queryScrollPage(scroll, scrollId1, size, resultClass);
            }
            @Override
            protected AppEsScrollPage<T> getFallback() {
                throw new AppEsFallbackException("scroll 查询失败");
            }
        };
        return command.execute();
    }



    /**
     * 关闭scroll请求
     * @param scrollId
     * @return
     */
    private boolean clearScrollRequest(String scrollId) {
        try {
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            //也可以选择setScrollIds()将多个scrollId一起使用
            clearScrollRequest.addScrollId(scrollId);
            ClearScrollResponse clearScrollResponse = restHighLevelClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
            return clearScrollResponse.isSucceeded();
        } catch (Exception e) {
            throw new AppEsUpdateException(e.getMessage());
        }
    }

    /**
     * 统计查询
     *
     * @param index              索引
     * @param queryBuilder       查询条件
     * @param aggregationBuilder
     * @return
     * @throws Exception
     */
    @Override
    public Aggregations aggregation(String index, QueryBuilder queryBuilder, AggregationBuilder aggregationBuilder) {
        HystrixCommand<Aggregations> command = new HystrixCommand<Aggregations>(getHystrixSetter("aggregation")){
            @Override
            protected Aggregations run() throws Exception {
                return appElasticSearchService.aggregation(index, queryBuilder, aggregationBuilder);
            }
            @Override
            protected Aggregations getFallback() {
                throw new AppEsFallbackException("统计查询失败");
            }
        };
        return command.execute();
    }

    /**
     * 使用分词查询,并分页
     *
     * @param startPage     当前页         从0开始
     * @param pageSize      每页显示条数
     * @param sortBuilders  排序字段
     * @return
     */
    @Override
    public AppEsPage<T> queryPage(String index, int startPage, int pageSize, QueryBuilder queryBuilder, List<SortBuilder> sortBuilders, Class<T> resultClass) {
        HystrixCommand<AppEsPage<T>> command = new HystrixCommand<AppEsPage<T>>(getHystrixSetter("queryPage")){
            @Override
            protected AppEsPage<T> run() throws Exception {
                return appElasticSearchService.queryPage(index, startPage, pageSize, queryBuilder, sortBuilders, resultClass);
            }
            @Override
            protected AppEsPage<T> getFallback() {
                throw new AppEsFallbackException("分页查询失败");
            }
        };
        return command.execute();
    }

    /**
     * 使用分词查询,并分页
     *
     * @param startPage     当前页            从0开始分页
     * @param pageSize      每页显示条数
     * @param sortBuilders  排序字段
     * @return
     */
    @Override
    public AppEsPage<Map> queryPage(String index, int startPage, int pageSize, QueryBuilder queryBuilder, List<SortBuilder> sortBuilders, String[] includeFields) {
        HystrixCommand<AppEsPage<Map>> command = new HystrixCommand<AppEsPage<Map>>(getHystrixSetter("queryPage1")){
            @Override
            protected AppEsPage<Map> run() throws Exception {
                return appElasticSearchService.queryPage(index, startPage, pageSize, queryBuilder, sortBuilders, includeFields);
            }
            @Override
            protected AppEsPage<Map> getFallback() {
                throw new AppEsFallbackException("分页查询失败");
            }
        };
        return command.execute();
    }

    /**
     * 删除单条数据
     */
    @Override
    public void deleteDocument(String index,  String id) {
        Assert.isTrue(UtilString.isNotEmpty(id), "id不能为空");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("deleteDocument")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.deleteDocument(index, id);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("删除数据失败");
            }
        };
        command.execute();
    }

    /**
     * 批量删除数据
     *
     * @param ids
     * @return
     * @throws IOException
     */
    @Override
    public void deleteBatchDocument(String index, List<String> ids) {
        Assert.isTrue(UtilCollection.isNotEmpty(ids), "id不能为空");

        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("deleteBatchDocument")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.deleteBatchDocument(index, ids);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("批量删除数据失败");
            }
        };
        command.execute();
    }


    /**
     * 公共批量bulk处理数据 插入、更新、删除
     */
    @Override
    public void bulkBatchDocument(BulkRequest bulkRequest) {
        Assert.isTrue(UtilCollection.isNotEmpty(bulkRequest.requests()), "数据不能为空");
        HystrixCommand<Boolean> command = new HystrixCommand<Boolean>(getHystrixSetter("bulkBatchDocument")){
            @Override
            protected Boolean run() throws Exception {
                appElasticSearchService.bulkBatchDocument(bulkRequest);
                return true;
            }
            @Override
            protected Boolean getFallback() {
                throw new AppEsFallbackException("批量处理数据失败");
            }
        };
        command.execute();
    }
}
